package edu.buffalo.cse.cse486_586.simpledynamo.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import android.R.integer;
import android.app.Service;
import android.content.ContentValues;
import android.content.Context;
import android.content.Intent;
import android.database.Cursor;
import android.net.Uri;
import android.os.Bundle;
import android.os.Handler;
import android.os.IBinder;
import android.os.Message;
import android.os.Messenger;
import android.os.RemoteException;
import android.telephony.TelephonyManager;
import android.util.Log;
import android.widget.Toast;
import edu.buffalo.cse.cse486_586.simpledynamo.entity.KeyValuePair;
import edu.buffalo.cse.cse486_586.simpledynamo.util.AppUtil;
import edu.buffalo.cse.cse486_586.simpledynamo.util.CommonUtil;

public class NetWorkService extends Service {
	// Debug params
	final private String TAG = "NetWorkService";
	// final private Boolean isDebug = true;
	// network listen thread
	Thread listenThread;
	ListeningThread listeningThread;
	boolean listenThreadFlag;
	// handler for communication between activity and service
	private Messenger outgoingMessager = null;
	private final Messenger incomingMessenger = new Messenger(
			new MessageHandler());
	// key-value pair container for buffer and commit
	private Map<String, KeyValuePair> bufferedKV;
	private Map<String, List<String>> ackPkgMap;
	// helpers
	private DynamoHelper dHelper;

	public NetWorkService() {
		super();
		this.dHelper = new DynamoHelper(this);
		this.listenThreadFlag = true;
		// init container
		this.bufferedKV = new HashMap<String, KeyValuePair>();
		this.ackPkgMap = new HashMap<String, List<String>>();
	}

	@Override
	public IBinder onBind(Intent intent) {
		intiPortConfig();
		listeningThread = new ListeningThread(this);
		listenThread = new Thread(listeningThread);
		listenThread.start();
		dHelper.sendRecoveryPkg();

		return incomingMessenger.getBinder();
	}

	@Override
	public void onDestroy() {
		try {
			super.onDestroy();
			Log.d(TAG, "onDestroy");
			listenThreadFlag = false;
			listeningThread.releaseSocket();
		} catch (IOException e) {
			Log.d(TAG, "exception", e);
		}
	}

	@Override
	public boolean onUnbind(Intent intent) {
		Log.d(TAG, "unbind");
		return true;
	}

	public void intiPortConfig() {
		int lp = getLocalConsolePort();
		dHelper.setLocalPort(lp);
		dHelper.setLocalRedirPort(AppUtil.getRedirPortByConsolePort(lp));
		dHelper.setLocalHashPort(CommonUtil.genHash(String.valueOf(lp)));
		dHelper.setNetWorkService(this);
	}

	/*
	 * if (the key-value pair belongs to the local machine) generate the unique
	 * version number and insert it else send it to targeted coorninator
	 */
	public void purposeValue(String key, String value) {
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));
		if (dHelper.getLocalRedirPort() == preferList.get(0)) {
			String vNum = dHelper.genVersionNumber();
			bufferKV(vNum, key, value);
			sendPreparePkg(vNum, key, value);
		} else {
			sendInsertPkg(key, value);
		}
	}

	/*
	 * commit the key-value pair into content provider if( key doesn't exist )
	 * commit else if( key exist && the new version number is greater than old
	 * version number ) commit else abort;
	 */
	public void commitKV(String key, String value, String vNum) {
		String hashKey = key;
		if (key.length() != 40)
			hashKey = CommonUtil.genHash(key);

		int oldVserion = isExist(hashKey);

		if (oldVserion == AppUtil.KEY_NONE_EXIST) {
			// insert new key-value
			ContentValues cv = new ContentValues();
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_KEY, hashKey);
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_VALUE, value);
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_VERSION, vNum);
			getContentResolver().insert(AppUtil.CONTENT_URI, cv);

		} else if (isUpdate(vNum, String.valueOf(oldVserion))) {
			ContentValues cv = new ContentValues();
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_KEY, hashKey);
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_VALUE, value);
			cv.put(AppUtil.KeyValueTable.COL_PROVIDER_VERSION, vNum);
			// uri, values, where, selectionArgs
			getContentResolver().update(AppUtil.CONTENT_URI, cv,
					AppUtil.KeyValueTable.COL_PROVIDER_KEY + "='" + hashKey + "'",
					null);
		}
	}

	public boolean isUpdate(String vNum1, String vNum2) {
		if (vNum1.length() > 5 && vNum2.length() > 5) {
			String console1 = vNum1.substring(0, 4);
			String console2 = vNum2.substring(0, 4);

			if (console1.equals(console2)
					&& Integer.valueOf(vNum1) > Integer.valueOf(vNum2))
				return true;
			else if (console1.equals(console2)
					&& Integer.valueOf(vNum1) < Integer.valueOf(vNum2))
				return false;
			else
				return true;
		} else {
			return true;
		}
	}

	/*
	 * Add prepare k-v pair into buffered Map
	 * 
	 * @return the version number (uuid) for the given k-v pair
	 */
	public String bufferKV(String vNum, String key, String value) {
		KeyValuePair kv = new KeyValuePair(key, value);
		bufferedKV.put(vNum, kv);
		return vNum;
	}

	/*
	 * return the local console port number
	 */
	public int getLocalConsolePort() {
		TelephonyManager tel = (TelephonyManager) getSystemService(Context.TELEPHONY_SERVICE);
		String portStr = tel.getLine1Number().substring(
				tel.getLine1Number().length() - 4);

		return Integer.parseInt(portStr);
	}

	public boolean isListenThreadFlag() {
		return listenThreadFlag;
	}

	public void setListenThreadFlag(boolean listenThreadFlag) {
		this.listenThreadFlag = listenThreadFlag;
	}

	/*
	 * check whether the key exist, key is hashed
	 * 
	 * @return the number of version
	 */
	public int isExist(String key) {
		Cursor c = getContentResolver().query(
				AppUtil.CONTENT_URI,
				new String[] { AppUtil.KeyValueTable.COL_ID,
						AppUtil.KeyValueTable.COL_PROVIDER_KEY,
						AppUtil.KeyValueTable.COL_PROVIDER_VALUE,
						AppUtil.KeyValueTable.COL_PROVIDER_VERSION },
				AppUtil.KeyValueTable.COL_PROVIDER_KEY + "='" + key + "'",
				null, null);

		if (c.moveToFirst()) {
			String version = c
					.getString(c
							.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_VERSION));
			if (CommonUtil.isNumeric(version))
				return Integer.parseInt(version);
			else
				return AppUtil.KEY_NONE_EXIST;
		} else {
			return AppUtil.KEY_NONE_EXIST;
		}
	}

	public void processReceivedContent(String pkg) {
		String tokens[] = pkg.split(AppUtil.PACKAGE_DELIMITER);
		// check whether the first token is number
		if (!CommonUtil.isNumeric(tokens[0]))
			return;
		int type = Integer.parseInt(tokens[0]);
		if (type >= AppUtil.PACKAGE_INSERT
				&& type <= AppUtil.PACKAGE_GET_RESPONSE) {
			processFunctionalPkg(pkg);
		} else {
			processFailurePkg(pkg);
		}
	}

	public boolean sendPkg(int redirPort, String pkg) {
		boolean isComplete = false;
		try {
			ExecutorService executor = Executors.newFixedThreadPool(1);
			Future<Boolean> futrue = executor.submit(new SendingThread(
					AppUtil.EMULATOR_IP, redirPort, pkg));
			isComplete = futrue.get();
		} catch (InterruptedException e) {
			Log.e(TAG, "Exception", e);
		} catch (ExecutionException e) {
			Log.e(TAG, "Exception", e);
		}

		return isComplete;
	}

	/*
	 * send insert package to the single cooperator
	 */
	public void sendInsertPkg(String key, String value) {
		String pkg = AppUtil.PACKAGE_INSERT + AppUtil.PACKAGE_DELIMITER + key
				+ AppUtil.PACKAGE_DELIMITER + value;

		// send preparing pkg to all replicas
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));
		if (preferList.size() > 0) {
			// handle failure
			// If there is a failure during sending Inserted message, send the
			// inserted message to the next coordinator
			for (int i = 0; i < preferList.size(); i++) {
				if (sendPkg(preferList.get(i), pkg)) {
					break;
				} else {
					String hashString = dHelper
							.getHasStringByRedirctPort(preferList.get(i));
					dHelper.changeGroupView(hashString, false);
					// boardcast failure
					dHelper.sendFailurePkg(hashString);
				}
			}
			// Thread t = new Thread(new SendingThread(AppUtil.EMULATOR_IP,
			// preferList.get(0), pkg));
			// t.start();
		}
	}

	/*
	 * send preparing pkg to all replicas
	 */
	public void sendPreparePkg(String vNum, String key, String value) {
		String pkg = AppUtil.PACKAGE_PREPARE + AppUtil.PACKAGE_DELIMITER + vNum
				+ AppUtil.PACKAGE_DELIMITER + key + AppUtil.PACKAGE_DELIMITER
				+ value;

		// send preparing pkg to all replicas
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));
		for (Integer port : preferList) {
			// handle failure
			// If there is a failure during sending prepare message,
			// just report the failur to all peers
			if (port != dHelper.getLocalRedirPort()) {
				boolean isComplete = sendPkg(port, pkg);
				if (!isComplete) {
					String hashString = dHelper.getHasStringByRedirctPort(port);
					dHelper.changeGroupView(hashString, false);
					// boardcast failure
					dHelper.sendFailurePkg(hashString);
				}
				// Thread t = new Thread(new SendingThread(AppUtil.EMULATOR_IP,
				// port, pkg));
				// t.start();
			}
		}
	}

	/*
	 * send ack pkg to all peers
	 */
	public void sendAckPackage(String vNum, String key, String value) {
		String pkg = AppUtil.PACKAGE_ACK + AppUtil.PACKAGE_DELIMITER + vNum
				+ AppUtil.PACKAGE_DELIMITER + key + AppUtil.PACKAGE_DELIMITER
				+ value;

		// send preparing pkg to all replicas
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));

		// handle failure
		// If there is a failure during sending ack message,
		// just report the failur to all peers
		for (Integer port : preferList) {
			if (port != dHelper.getLocalRedirPort()) {
				if (!sendPkg(port, pkg)) {
					String hashString = dHelper.getHasStringByRedirctPort(port);
					dHelper.changeGroupView(hashString, false);
					// boardcast failure
					dHelper.sendFailurePkg(hashString);
				} else {
					break;
				}
				// Thread t = new Thread(new SendingThread(AppUtil.EMULATOR_IP,
				// port, pkg));
				// t.start();
			}
		}
	}

	/*
	 * send commit package to replicas
	 */
	public void sendCommitPkg(String uuid, String key, String value) {
		String pkg = AppUtil.PACKAGE_COMMIT + AppUtil.PACKAGE_DELIMITER + uuid;

		// send preparing pkg to all replicas
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));
		// handle failure
		// If there is a failure during sending commit message,
		// just report the failur to all peers
		for (Integer port : preferList) {
			if (port != dHelper.getLocalRedirPort()) {
				if (!sendPkg(port, pkg)) {
					String hashString = dHelper.getHasStringByRedirctPort(port);
					dHelper.changeGroupView(hashString, false);
					// boardcast failure
					dHelper.sendFailurePkg(hashString);
				}
				// Thread t = new Thread(new SendingThread(AppUtil.EMULATOR_IP,
				// port, pkg));
				// t.start();
			}
		}
	}

	// PACKAGE_INSERT format: 1/--/key/--/value ;
	// PACKAGE_PREPARE format: 2/--/versionNum/--/key/--/value ;
	// PACKAGE_ACK format: 3/--/versionNum/--/key/--/value ;
	// PACKAGE_COMMIT format: 4/--/versionNum ;
	public void processFunctionalPkg(String pkg) {
		String tokens[] = pkg.split(AppUtil.PACKAGE_DELIMITER);
		// check whether the first token is number
		if (!CommonUtil.isNumeric(tokens[0]))
			return;
		int type = Integer.parseInt(tokens[0]);
		switch (type) {
		case AppUtil.PACKAGE_INSERT:
			purposeValue(tokens[1], tokens[2]);
			break;
		case AppUtil.PACKAGE_PREPARE:
			bufferKV(tokens[1], tokens[2], tokens[3]);
			sendAckPackage(tokens[1], tokens[2], tokens[3]);
			break;
		case AppUtil.PACKAGE_ACK:
			String vNum = tokens[1];
			String key = tokens[2];
			if (ackPkgMap.containsKey(vNum)) {
				ackPkgMap.get(vNum).add(pkg);
			} else {
				List<String> l = new ArrayList<String>();
				l.add(pkg);
				ackPkgMap.put(vNum, l);
			}
			List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
					.genHash(key));
			if (ackPkgMap.get(vNum).size() >= (preferList.size() - 1)) {
				KeyValuePair kv = bufferedKV.get(vNum);
				commitKV(kv.getKey(), kv.getValue(), vNum);
				// send commit message to replica
				sendCommitPkg(vNum, kv.getKey(), kv.getValue());
				// remove the instance in buffered container and ack list
				bufferedKV.remove(vNum);
				ackPkgMap.remove(vNum);
			}
			break;
		case AppUtil.PACKAGE_COMMIT:
			String versionNumber = tokens[1];
			if (bufferedKV.containsKey(versionNumber)) {
				KeyValuePair kv = bufferedKV.get(versionNumber);
				commitKV(kv.getKey(), kv.getValue(), versionNumber);
				// remove the instance in buffered container
				bufferedKV.remove(versionNumber);
			}
			break;
		case AppUtil.PACKAGE_GET:
			String displayStr = searchFromCP(tokens[2]);
			sendGetResponsePkg(tokens[1], displayStr);
			break;
		case AppUtil.PACKAGE_GET_RESPONSE:
			String str = tokens[1];
			// send response to activity
			Map<String, String> map = new HashMap<String, String>();
			map.put(AppUtil.DISPLAY_RESULT, str);
			sendMessageToActivity(AppUtil.MSG_UPDATE_VIEWLIST, map);
			break;
		default:
			break;
		}
	}

	// PACKAGE_FAILUR_REPORT format : 7/--/console port hashed string
	// PACKAGE_RECOVERY_REPORT format : 8/--/console port hashed string
	public void processFailurePkg(String pkg) {
		String tokens[] = pkg.split(AppUtil.PACKAGE_DELIMITER);
		if (!CommonUtil.isNumeric(tokens[0]))
			return;
		int type = Integer.parseInt(tokens[0]);
		switch (type) {
		case AppUtil.PACKAGE_FAILUR_REPORT:
			if (dHelper.isAlive(tokens[1]))
				dHelper.changeGroupView(tokens[1], false);
			break;
		case AppUtil.PACKAGE_RECOVERY_REPORT:
			if (!dHelper.isAlive(tokens[1])) {
				Log.d(TAG, "chane group view");
				dHelper.changeGroupView(tokens[1], true);
				// send the kv need to synchronized
				String content = responseForRecovery(tokens[1]);
				Log.d(TAG, "respone kv list ===>" + content);
				if (content.trim().length() > 0)
					dHelper.sendSynPkg(tokens[1], AppUtil.PACKAGE_SYN
							+ AppUtil.PACKAGE_DELIMITER + content);
			}
			break;
		case AppUtil.PACKAGE_SYN:
			if (tokens.length > 1) {
				String kvArray[] = tokens[1].trim().split(
						AppUtil.KV_LIST_DELIMITER);
				for (int i = 0; i < kvArray.length; i++) {
					String attrs[] = kvArray[i].trim().split(
							AppUtil.KV_ATTR_DELIMITER);
					if (attrs.length == 3) {
						commitKV(attrs[0], attrs[1], attrs[2]);
					}
				}
			}
			break;
		default:
			break;
		}
	}

	/*
	 * retrieve key-value pair
	 */
	private void retreiveKvByKey(String key) {
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));
		if (preferList.size() > 0) {
			if (preferList.get(0) == dHelper.getLocalRedirPort()) {
				// retieve from local storage
				String kvStr = searchFromCP(key);
				// send display event back to activity
				Map<String, String> map = new HashMap<String, String>();
				map.put(AppUtil.DISPLAY_RESULT, kvStr);
				sendMessageToActivity(AppUtil.MSG_UPDATE_VIEWLIST, map);
			} else {
				// send retrive message to single coordinator
				sendGetPkg(key);
			}
		}
	}

	public String searchFromCP(String key) {
		String result = "";
		Cursor c = getContentResolver().query(
				AppUtil.CONTENT_URI,
				new String[] { AppUtil.KeyValueTable.COL_ID,
						AppUtil.KeyValueTable.COL_PROVIDER_KEY,
						AppUtil.KeyValueTable.COL_PROVIDER_VALUE,
						AppUtil.KeyValueTable.COL_PROVIDER_VERSION },
				AppUtil.KeyValueTable.COL_PROVIDER_KEY + "='"
						+ CommonUtil.genHash(key) + "'", null, null);

		if (c.moveToFirst()) {
			String hashKey = c
					.getString(c
							.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_KEY));
			String value = c
					.getString(c
							.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_VALUE));
			String version = c
					.getString(c
							.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_VERSION));

			result = "key : " + hashKey + "\n value : " + value
					+ "\n version : " + version;
		}

		return result;
	}

	public void sendGetPkg(String key) {
		String pkg = AppUtil.PACKAGE_GET + AppUtil.PACKAGE_DELIMITER
				+ dHelper.getLocalRedirPort() + AppUtil.PACKAGE_DELIMITER + key;

		// send preparing pkg to all replicas
		List<Integer> preferList = dHelper.getPreferenceList(CommonUtil
				.genHash(key));

		// handle failure
		// If there is a failure during sending get message, send the
		// get message to the next coordinator
		for (int i = 0; i < preferList.size(); i++) {
			if (preferList.get(i) != dHelper.getLocalRedirPort()) {
				if (sendPkg(preferList.get(i), pkg)) {
					break;
				} else {
					String hashString = dHelper
							.getHasStringByRedirctPort(preferList.get(i));
					dHelper.changeGroupView(hashString, false);
					// boardcast failure
					dHelper.sendFailurePkg(hashString);
				}
			}
		}

	}

	public String responseForRecovery(String recoveredHashPort) {
		Cursor c = getContentResolver().query(
				AppUtil.CONTENT_URI,
				new String[] { AppUtil.KeyValueTable.COL_ID,
						AppUtil.KeyValueTable.COL_PROVIDER_KEY,
						AppUtil.KeyValueTable.COL_PROVIDER_VALUE,
						AppUtil.KeyValueTable.COL_PROVIDER_VERSION }, null,
				null, null);

		String pkg = "";
		if (c.moveToFirst()) {
			while (!c.isAfterLast()) {
				String key = c
						.getString(c
								.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_KEY));
				String value = c
						.getString(c
								.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_VALUE));
				String version = c
						.getString(c
								.getColumnIndexOrThrow(AppUtil.KeyValueTable.COL_PROVIDER_VERSION));

				int kvType = dHelper.getKvType(key, recoveredHashPort);
				if (kvType == AppUtil.KV_LOCAL_STORAGE
						|| kvType == AppUtil.KV_REPLICA) {
					pkg += key + AppUtil.KV_ATTR_DELIMITER + value
							+ AppUtil.KV_ATTR_DELIMITER + version
							+ AppUtil.KV_LIST_DELIMITER;
				}

				c.moveToNext();
			}
		}

		return pkg;
	}

	public void sendGetResponsePkg(String redirPort, String str) {
		String pkg = AppUtil.PACKAGE_GET_RESPONSE + AppUtil.PACKAGE_DELIMITER
				+ str;

		if (!sendPkg(Integer.parseInt(redirPort), pkg)) {
			String hashString = dHelper.getHasStringByRedirctPort(Integer
					.parseInt(redirPort));
			dHelper.changeGroupView(hashString, false);
			// boardcast failure message
			dHelper.sendFailurePkg(hashString);
		}

		// Thread t = new Thread(new SendingThread(AppUtil.EMULATOR_IP,
		// Integer.parseInt(redirPort), pkg));
		// t.start();
	}

	/*
	 * dispatch the event to activity
	 */
	public void sendMessageToActivity(int message, Map<String, String> data) {
		try {
			Message msg = Message.obtain(null, message);
			android.os.Bundle b = new android.os.Bundle();

			if (data != null) {
				for (Map.Entry<String, String> item : data.entrySet()) {
					b.putString(item.getKey(), item.getValue());
				}
			}
			msg.setData(b);
			outgoingMessager.send(msg);
		} catch (RemoteException e) {
			Log.e(TAG, "Exception", e);
		}
	}

	// Handler of messages from binded activity.
	class MessageHandler extends Handler {
		@Override
		public void handleMessage(Message msg) {
			switch (msg.what) {
			case AppUtil.MSG_REGISTER_CLIENT:
				outgoingMessager = msg.replyTo;
				break;
			case AppUtil.MSG_PUT_CASE:
				Bundle b = msg.getData();
				final String putPrefix = b.getString(AppUtil.PUT_PRIFIX);
				Thread t = new Thread(new Runnable() {
					@Override
					public void run() {
						for (int i = 0; i < 10; i++) {
							try {
								Thread.sleep(3000);
								Map<String, String> map = new HashMap<String, String>();
								map.put(AppUtil.DISPLAY_RESULT, "Insert key : " + i + ", putPrefix" + i);
								sendMessageToActivity(AppUtil.MSG_DISPLAY_TOAST, map);
								purposeValue(String.valueOf(i), putPrefix + i);
							} catch (InterruptedException e) {
								Log.e(TAG, "Exception", e);
							}
						}
					}
				});
				t.start();
				break;
			case AppUtil.MSG_GET_CASE:
				Thread t1 = new Thread(new Runnable() {
					@Override
					public void run() {
						for (int i = 0; i < 10; i++) {
							try {
								Thread.sleep(3000);
								retreiveKvByKey(String.valueOf(i));
							} catch (InterruptedException e) {
								Log.e(TAG, "Exception", e);
							}
						}
					}
				});
				t1.start();
				break;
			case AppUtil.MSG_ADD:
				break;
			default:
				super.handleMessage(msg);
			}
		}
	}
}
